package org.codehaus.activemq.service.impl;

import javax.jms.JMSException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.message.ActiveMQMessage;
import org.codehaus.activemq.message.MessageAck;
import org.codehaus.activemq.service.MessageIdentity;
import org.codehaus.activemq.service.Subscription;
import org.codehaus.activemq.service.TopicMessageContainer;
import org.codehaus.activemq.store.TopicMessageStore;

public class DurableTopicMessageContainer
  implements TopicMessageContainer
{
  private static final Log log = LogFactory.getLog(DurableTopicMessageContainer.class);
  private TopicMessageStore messageStore;
  private String destinationName;
  private MessageIdentity lastMessageIdentity;

  public DurableTopicMessageContainer(TopicMessageStore messageStore, String destinationName)
  {
    this.messageStore = messageStore;
    this.destinationName = destinationName;
  }

  public String getDestinationName() {
    return this.destinationName;
  }

  public MessageIdentity addMessage(ActiveMQMessage message) throws JMSException {
    MessageIdentity answer = this.messageStore.addMessage(message);
    this.lastMessageIdentity = answer;
    return answer;
  }

  public void delete(MessageIdentity messageID, MessageAck ack)
    throws JMSException
  {
  }

  public boolean containsMessage(MessageIdentity messageIdentity) throws JMSException
  {
    return getMessage(messageIdentity) != null;
  }

  public ActiveMQMessage getMessage(MessageIdentity messageID) throws JMSException {
    return this.messageStore.getMessage(messageID);
  }

  public void registerMessageInterest(MessageIdentity messageIdentity) throws JMSException {
    this.messageStore.incrementMessageCount(messageIdentity);
  }

  public void unregisterMessageInterest(MessageIdentity messageIdentity, MessageAck ack) throws JMSException {
    this.messageStore.decrementMessageCountAndMaybeDelete(messageIdentity, ack);
  }

  public void setLastAcknowledgedMessageID(Subscription subscription, MessageIdentity messageIdentity) throws JMSException
  {
    this.messageStore.setLastAcknowledgedMessageIdentity(subscription, messageIdentity);
  }

  public void recoverSubscription(Subscription subscription) throws JMSException {
    this.messageStore.recoverSubscription(subscription, this.lastMessageIdentity);
  }

  public void start() throws JMSException
  {
    this.messageStore.setMessageContainer(this);
    this.lastMessageIdentity = this.messageStore.getLastestMessageIdentity();
    this.messageStore.start();
  }

  public void stop() throws JMSException {
    this.messageStore.stop();
  }
}